/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.broker.topic;

import com.google.common.collect.Sets;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;

/**
 * 主题路由信息管理器
 */
public class TopicRouteInfoManager {

    /**
     * 获取主题路由信息的超时时间，因为需要请求nameserver。
     * 默认值：3秒
     */
    private static final long GET_TOPIC_ROUTE_TIMEOUT = 3000L;

    /**
     * 主题路由信息锁定超时时间。
     * 默认值：3秒
     */
    private static final long LOCK_TIMEOUT_MILLIS = 3000L;

    private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    /**
     * 互斥锁
     */
    private final Lock lockNamesrv = new ReentrantLock();

    /**
     * key为主题名称，value为主题路由数据组成的ConcurrentHashMap集合。
     */
    private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<>();

    /**
     * key为broker名称，一个broker名称对应broker的主从集群；
     * value为主从集群，key为brokerId，value为broker地址。
     */
    private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
        new ConcurrentHashMap<>();

    /**
     * 记录主题以及主题路由信息的集合。
     * key是主题名称，value是主题发布信息对象。
     *
     * 主题发布信息包含如下内容：
     * <ol>
     *     <li>是否是有序主题，默认为false</li>
     *     <li>是否包含主题路由信息，默认为false</li>
     *     <li>消息队列List集合</li>
     *     <li>用于生成和保存当前线程挑选队列的索引的对象</li>
     *     <li>主题路由数据，用于记录主题路由信息的类</li>
     * </ol>
     */
    private final ConcurrentMap<String/* topic */, TopicPublishInfo> topicPublishInfoTable = new ConcurrentHashMap<>();

    /**
     *
     */
    private final ConcurrentHashMap<String, Set<MessageQueue>> topicSubscribeInfoTable = new ConcurrentHashMap<>();

    /**
     * 用于执行调度任务的线程池
     */
    private ScheduledExecutorService scheduledExecutorService;

    /**
     * broker控制器
     */
    private BrokerController brokerController;

    public TopicRouteInfoManager(BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    /**
     * 启动主题路由信息管理器，使用调度线程池周期性地使用nameserver的元数据更新本地缓存元数据的操作。
     */
    public void start() {
        // 只包含一个线程的调度线程池
        this.scheduledExecutorService = ThreadUtils.newSingleThreadScheduledExecutor(
                new ThreadFactoryImpl("TopicRouteInfoManagerScheduledThread")
        );

        // 以固定频率调度执行任务，延迟1秒开始执行，默认每隔30秒执行一次。
        this.scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                // 根据NameServer的数据更新主题路由信息
                updateTopicRouteInfoFromNameServer();
            } catch (Exception e) {
                log.error("ScheduledTask: failed to pull TopicRouteData from NameServer", e);
            }
        }, 1000, this.brokerController.getBrokerConfig().getLoadBalancePollNameServerInterval(), TimeUnit.MILLISECONDS);
    }

    /**
     * 使用从nameserver拉取过来的元数据更新主题路由信息
     */
    private void updateTopicRouteInfoFromNameServer() {
        // 订阅信息表
        final Set<String> topicSetForPopAssignment = this.topicSubscribeInfoTable.keySet();
        // 主题路由表
        final Set<String> topicSetForEscapeBridge = this.topicRouteTable.keySet();
        // 使用union方法生成遍历两个集合的视图，首先遍历第一个集合中的元素，遍历结束后，再接着遍历集合2中包含但不在集合1中的元素。
        final Set<String> topicsAll = Sets.union(topicSetForPopAssignment, topicSetForEscapeBridge);

        // 遍历topicsAll中的主题名称
        for (String topic : topicsAll) {
            // 是否需要更新主题发布信息？
            boolean isNeedUpdatePublishInfo = topicSetForEscapeBridge.contains(topic);
            // 是否需要更新主题订阅信息
            boolean isNeedUpdateSubscribeInfo = topicSetForPopAssignment.contains(topic);
            // 对当前主题，执行更新
            updateTopicRouteInfoFromNameServer(topic, isNeedUpdatePublishInfo, isNeedUpdateSubscribeInfo);
        }
    }

    /**
     * 从NameServer获取主题路由信息，并更新本地主题路由信息缓存。
     * @param topic 主题名称
     * @param isNeedUpdatePublishInfo  是否需要更新【】
     * @param isNeedUpdateSubscribeInfo 是否需要更新【订阅信息表】
     */
    public void updateTopicRouteInfoFromNameServer(String topic, boolean isNeedUpdatePublishInfo,
        boolean isNeedUpdateSubscribeInfo) {
        try {
            // 获取互斥锁，设置锁定的超时时间
            if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    // 向NameServer发起关于查询指定主题路由信息的请求，并获取结果。
                    // 注意，这里设置了请求的超时时间
                    final TopicRouteData topicRouteData = this.brokerController.getBrokerOuterAPI()
                        .getTopicRouteInfoFromNameServer(topic, GET_TOPIC_ROUTE_TIMEOUT);
                    // 如果没有获取到，则方法返回，更新失败。
                    if (null == topicRouteData) {
                        log.warn("TopicRouteInfoManager: updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}.", topic);
                        return;
                    }

                    // 如果需要更新订阅信息表，则使用nameserver的数据更新订阅数据表
                    if (isNeedUpdateSubscribeInfo) {
                        this.updateSubscribeInfoTable(topicRouteData, topic);
                    }

                    // 如果需要更新主题路由信息
                    if (isNeedUpdatePublishInfo) {
                        // 则更新主题路由信息
                        this.updateTopicRouteTable(topic, topicRouteData);
                    }
                } catch (RemotingException e) {
                    log.error("updateTopicRouteInfoFromNameServer Exception", e);
                } catch (MQBrokerException e) {
                    log.error("updateTopicRouteInfoFromNameServer Exception", e);
                    if (!NamespaceUtil.isRetryTopic(topic)
                        && ResponseCode.TOPIC_NOT_EXIST == e.getResponseCode()) {
                        // clean no used topic
                        cleanNoneRouteTopic(topic);
                    }
                } finally {
                    // 操作结束后释放互斥锁
                    this.lockNamesrv.unlock();
                }
            }
        } catch (InterruptedException e) {
            log.warn("updateTopicRouteInfoFromNameServer Exception", e);
        }
    }

    /**
     * 更新主题路由信息表
     * @param topic
     * @param topicRouteData
     * @return
     */
    private boolean updateTopicRouteTable(String topic, TopicRouteData topicRouteData) {
        // 从主题路由表中获取指定主题当前缓存的路由信息
        TopicRouteData old = this.topicRouteTable.get(topic);
        // 判断路由信息是否发生变更
        boolean changed = topicRouteData.topicRouteDataChanged(old);
        // 如果没有发生变更，并且也不需要更新主题路由信息，则返回false。
        if (!changed) {
            if (!this.isNeedUpdateTopicRouteInfo(topic)) {
                return false;
            }
        } else {
            log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);
        }

        // 如果需要更新
        for (BrokerData bd : topicRouteData.getBrokerDatas()) {
            // broker地址表添加条目
            this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
        }

        // 将主题路由信息对象转换为主题发布信息对象
        TopicPublishInfo publishInfo = MQClientInstance.topicRouteData2TopicPublishInfo(topic, topicRouteData);
        // 设置包含主题路由信息为true
        publishInfo.setHaveTopicRouterInfo(true);


        // 更新主题的发布信息
        this.updateTopicPublishInfo(topic, publishInfo);


        // 创建主题路由数据对象
        TopicRouteData cloneTopicRouteData = new TopicRouteData(topicRouteData);
        log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);
        // 将主题路由数据对象放入主题路由表中
        this.topicRouteTable.put(topic, cloneTopicRouteData);

        // 返回true
        return true;
    }

    /**
     * 更新订阅信息表
     * @param topicRouteData 主题路由数据
     * @param topic          主题名称
     * @return
     */
    private boolean updateSubscribeInfoTable(TopicRouteData topicRouteData, String topic) {
        // 根据主题路由数据创建主题路由数据
        final TopicRouteData tmp = new TopicRouteData(topicRouteData);
        // 按照broker进行主题队列映射的数据设置为null
        tmp.setTopicQueueMappingByBroker(null);
        // 将主题路由数据转换为主题订阅信息（订阅信息是消息队列的集合）
        Set<MessageQueue> newSubscribeInfo = MQClientInstance.topicRouteData2TopicSubscribeInfo(topic, tmp);
        // 从主题订阅信息表中获取指定主题的订阅信息
        Set<MessageQueue> oldSubscribeInfo = topicSubscribeInfoTable.get(topic);

        // 如果缓存的主题订阅信息与指定的主题订阅信息相同，则返回false，表示订阅数据无需更新。
        if (Objects.equals(newSubscribeInfo, oldSubscribeInfo)) {
            return false;
        }

        // 否则，将新的主题订阅信息添加到主题订阅信息表中，替换原来的主题订阅信息
        log.info("the topic[{}] subscribe message queue changed, old[{}] ,new[{}]", topic, oldSubscribeInfo, newSubscribeInfo);
        topicSubscribeInfoTable.put(topic, newSubscribeInfo);
        return true;

    }

    /**
     * 是否需要更新主题路由信息：
     * -
     * @param topic
     * @return
     */
    private boolean isNeedUpdateTopicRouteInfo(final String topic) {
        // 从缓存中获取指定主题的路由信息
        final TopicPublishInfo prev = this.topicPublishInfoTable.get(topic);
        // 如果是null值，则需要更新主题路由信息。如果数据没有准备好，则也需要更新主题路由信息
        return null == prev || !prev.ok();
    }

    /**
     * 将没有路由信息的主题从主题订阅信息表中移除。
     * @param topic 没有主题路由信息的主题名称
     */
    private void cleanNoneRouteTopic(String topic) {
        // clean no used topic
        topicSubscribeInfoTable.remove(topic);
    }

    /**
     * 更新主题的发布信息
     * @param topic 需要更新发布信息的主题名称
     * @param info  主题的新的发布信息
     */
    private void updateTopicPublishInfo(final String topic, final TopicPublishInfo info) {
        // 如果主题名称和主题发布信息都存在，则直接以主题名称为key，主题发布信息为value，添加到主题发布信息表中。
        if (info != null && topic != null) {
            TopicPublishInfo prev = this.topicPublishInfoTable.put(topic, info);
            // 如果prev不是null，表示新的主题发布信息替换了原来的主题发布信息
            if (prev != null) {
                log.info("updateTopicPublishInfo prev is not null, " + prev);
            }
        }
    }

    /**
     * 关闭主题路由信息管理器
     */
    public void shutdown() {
        // 如果调度线程池服务不是null，则关闭调度线程池服务
        // 该调度线程池服务就是主题路由信息管理器的主要工作，关闭了调度线程池服务，也意味着关闭了主题路由信息服务。
        if (null != this.scheduledExecutorService) {
            this.scheduledExecutorService.shutdown();
        }
    }

    /**
     * 尝试查找主题发布信息：
     * 1. 从本地缓存中查找指定主题的主题发布信息；
     * 2. 如果本地没有，则请求NameServer，更新本地缓存信息；
     * 3. 从本地缓存中查找指定主题的主题发布信息并返回。
     *
     * @param topic 需要查找主题发布信息的主题名称
     * @return
     */
    public TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
        // 从主题发布信息表中获取指定主题的主题发布信息
        TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
        // 如果没有找到，或找到了但是没有数据，则
        if (null == topicPublishInfo || !topicPublishInfo.ok()) {
            // 向nameserver发起请求，查询主题的路由信息，并更新主题发布信息表中的主题发布信息
            this.updateTopicRouteInfoFromNameServer(topic, true, false);
            // 从主题发布信息表中获取指定主题的发布信息
            topicPublishInfo = this.topicPublishInfoTable.get(topic);
        }
        // 返回主题发布信息
        return topicPublishInfo;
    }

    /**
     * 根据broker名称查找broker的地址：
     * - 由于一个broker名称对应一组组成主从关系的broker信息；
     * - 如果查找到的集合不是null，不是空，则直接获取master节点的地址并返回；
     * - 否则直接返回null。
     * @param brokerName broker名称
     * @return 返回查找到的broker master地址
     */
    public String findBrokerAddressInPublish(String brokerName) {
        // 如果主题名称为null，则直接返回null。
        if (brokerName == null) {
            return null;
        }

        // 从broker地址表中查找指定broker名称的地址信息（注意：一个broker名称对应组成主从集群的多个broker的id和地址信息）
        Map<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
        // 如果不是null，不是空，则直接获取master节点的地址并返回。
        if (map != null && !map.isEmpty()) {
            return map.get(MixAll.MASTER_ID);
        }

        // 否则返回null。
        return null;
    }

    /**
     * 根据broker名称和brokerId查找broker的地址
     * @param brokerName broker名称
     * @param brokerId   broker的ID
     * @param onlyThisBroker 是否在当前broker上查找，true：在当前broker上查找，false：在所有broker上查找
     * @return 返回找到的broker地址
     */
    public String findBrokerAddressInSubscribe(
        final String brokerName, final long brokerId, final boolean onlyThisBroker
    ) {

        // 如果broker名称为null，则返回null。
        if (brokerName == null) {
            return null;
        }

        // 用于存储找到的broker地址
        String brokerAddr = null;
        // 找到的标志，true表示找到
        boolean found = false;

        // 从broker地址表中获取指定broker名称的broker集群地址信息
        Map<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);

        // 如果不是null，不是空，则
        if (map != null && !map.isEmpty()) {
            // 获取指定brokerId的broker地址
            brokerAddr = map.get(brokerId);
            // 是否是slave的标志：只有brokerId是0时才表示master节点
            boolean slave = brokerId != MixAll.MASTER_ID;

            // 设置查找结果的标志
            found = brokerAddr != null;

            // 如果没有找到该slave，就查找下一个slave
            if (!found && slave) {
                // 则查找下一个slave
                brokerAddr = map.get(brokerId + 1);
                // 设置查找结果的标志
                found = brokerAddr != null;
            }

            // 如果还是没有找到这个slave，且不仅是在当前broker上查找，则找迭代器获取到的第一个节点信息
            if (!found && !onlyThisBroker) {
                Map.Entry<Long, String> entry = map.entrySet().iterator().next();
                brokerAddr = entry.getValue();
                found = true;
            }
        }
        // 返回broker地址
        return brokerAddr;
    }

    /**
     * 根据主题名称查找主题的消息队列集合。
     * @param topic 主题名称
     * @return 返回找到的指定主题的消息队列集合。
     */
    public Set<MessageQueue> getTopicSubscribeInfo(String topic) {
        // 从主题订阅信息表中获取指定主题的消息队列集合
        Set<MessageQueue> queues = topicSubscribeInfoTable.get(topic);
        if (null == queues || queues.isEmpty()) {
            // 向nameserver发起请求更新本地主题路由信息
            this.updateTopicRouteInfoFromNameServer(topic, false, true);
            // 重新获取指定主题的消息队列集合
            queues = this.topicSubscribeInfoTable.get(topic);
        }
        // 返回找到的消息队列集合
        return queues;
    }
}
